/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.zookeeper.server.quorum;

import java.util.ArrayList;
import java.util.LinkedList;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.zookeeper.ZooDefs.OpCode;
import org.apache.zookeeper.server.Request;
import org.apache.zookeeper.server.RequestProcessor;

/**
 * 事务提交处理器。对于非事务请求，该处理器会直接将其交付给下一级处理器处理；
 * 对于事务请求，其会等待集群内针对Proposal的投票直到该Proposal可被提交，利用CommitProcessor，每个服务器都可以很好地控制对事务请求的顺序处理。
 */
public class CommitProcessor extends Thread implements RequestProcessor {
    private static final Logger LOG = LoggerFactory.getLogger(CommitProcessor.class);

    /** 请求队列, 接收到的需要进行等待commit的请求以及不需要等待的非事务请求 */
    LinkedList<Request> queuedRequests = new LinkedList<>();
    /** 这个队列存储的是已经commit的请求,不过也有可能有其他服接收的请求由leader发送到follwer进行commit的操作。 */
    LinkedList<Request> committedRequests = new LinkedList<>();

    RequestProcessor nextProcessor;
    /** 待处理的队列, 要交付给下一个处理器进行后续处理 */
    ArrayList<Request> toProcess = new ArrayList<>();

    /**
     * 看sync的请求是等待leader回复，还是说直接处理，像读请求一样。对于leader是false，对于learner是true
     */
    boolean matchSyncs;

    public CommitProcessor(RequestProcessor nextProcessor, String id, boolean matchSyncs) {
        super("CommitProcessor:" + id);
        this.nextProcessor = nextProcessor;
        this.matchSyncs = matchSyncs;
    }

    volatile boolean finished = false;

    /**
     * 这是一个单线程，然后利用一个无线循环来进行操作，每次循环主要是进行三块操作
     * 1. 将toProcess中的请求包交给后续处理器处理
     * 2. 处理commitedRequests的数据
     * 3. 处理queuedRequests中的数据
     */
    @Override
    public void run() {
        try {
            // 下一个未处理的事务请求(不含leader端的sync请求),只要为null，都会while循环从queuedRequests里面找到第一个事务请求，或者直到队列为空
            Request nextPending = null;
            // 只要没有shutdown
            while (!finished) {
                // 下面将 committedRequests 封装后 加入 toProcess,
                int len = toProcess.size();
                for (int i = 0; i < len; i++) {
                    // 待处理队列交给下个处理器 FinalRequestProcessor,按顺序处理
                    nextProcessor.processRequest(toProcess.get(i));
                }
                //队列清空
                toProcess.clear();
                //注意这里上锁，不会出现执行到过程中，queuedRequests的size变了
                synchronized (this) {
                    //这部分结合尾部的while来读，要么 请求队列remove干净，要么从中找到一个事务请求，赋值给nextPending, 不允许size>0 且nextPending == null的情况
                    if ((queuedRequests.size() == 0 || nextPending != null)
                                                && committedRequests.size() == 0) { // 且 没有已提交事务
                        wait();
                        continue;
                    }

                    // 这里其他的服务器接收的事务请求的优先级要低于本地接收的非事务请求
                    // 不允许size>0且nextPending == null的情况
                    if ((queuedRequests.size() == 0 || nextPending != null)
                                        && committedRequests.size() > 0) { //如果有 已提交的请求
                        Request r = committedRequests.remove();

                        // 如果和nextPending匹配
                        // 当前等待包与commit的包是一样的sessionId和cxid,则表示的是当前正在等待的commit的包
                        if (nextPending != null && nextPending.sessionId == r.sessionId && nextPending.cxid == r.cxid) {
                            nextPending.hdr = r.hdr;
                            nextPending.txn = r.txn;
                            nextPending.zxid = r.zxid;
                            //加入待处理队列
                            toProcess.add(nextPending);
                            //下一个pend的请求清空
                            nextPending = null;
                        } else { // 其他的服务器接收的包直接进行后续的处理
                            // 这种情况是nextPending还没有来的及设置，nextPending==null的情况(代码应该再细分一下if else),不可能出现nextPending!=null而走到了这里的情况(算异常)
                            toProcess.add(r);
                        }
                    }
                }
                // 如果还有 未处理的事务请求(不含leader端的sync请求),就continue
                // 当前还有正在等待commit的包，则不进行后续的对queuedRequests的处理
                if (nextPending != null) {
                    continue;
                }

                // 这一段的目的是找到一个 给nextPending赋值
                synchronized (this) { //处理请求包
                    //只要queuedRequests队列不空，从中找到第一个 事务请求(不含leader端的sync请求),前面的其他请求全部加入待处理队列
                    while (nextPending == null && queuedRequests.size() > 0) {
                        Request request = queuedRequests.remove();
                        switch (request.type) {
                        case OpCode.create:
                        case OpCode.delete:
                        case OpCode.setData:
                        case OpCode.multi:
                        case OpCode.setACL:
                        case OpCode.createSession:
                        case OpCode.closeSession:
                            nextPending = request; // 事务请求使用nextPending等待
                            break; //大部分事务请求直接赋给nextPending，然后break
                        case OpCode.sync:
                            if (matchSyncs) { //如果需要等leader返回,该值learner端为true
                                nextPending = request;
                            } else {
                                toProcess.add(request); //不需要的话，直接加入待处理队列里
                            }
                            break; //leader端matchSyncs是false，learner端才需要等leader回复，这里也break
                        default:
                            toProcess.add(request); //非事务请求，都直接加入待处理队列
                        }
                    }
                }
            }
        } catch (InterruptedException e) {
            LOG.warn("Interrupted exception while waiting", e);
        } catch (Throwable e) {
            LOG.error("Unexpected exception causing CommitProcessor to exit", e);
        }
        LOG.info("CommitProcessor exited loop!");
    }

    /**
     * 事务请求提交
     * @param request
     */
    synchronized public void commit(Request request) {
        //只要没有结束
        if (!finished) {
            if (request == null) {
                LOG.warn("Committed a null!", new Exception("committing a null! "));
                return;
            }
            //进入已提交队列
            committedRequests.add(request);
            //通知
            notifyAll();
        }
    }

    @Override
    synchronized public void processRequest(Request request) {
        if (!finished) {
            queuedRequests.add(request);
            notifyAll();
        }
    }

    @Override
    public void shutdown() {
        LOG.info("Shutting down");
        synchronized (this) {
            finished = true;
            queuedRequests.clear();
            notifyAll();
        }
        if (nextProcessor != null) {
            nextProcessor.shutdown();
        }
    }

}
